package com.ibm.util;

import java.util.Vector;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.kafka.KafkaProducer;
import com.ibm.cps.message.MessageFactory.MessageType;
import com.ibm.cps.message.MessageFields;
import com.ibm.util.exception.CPSException;

public class FeedbackMessage {

	private static KafkaProducer producer = new KafkaProducer(
			LocalConfig.KAFKA_BROKERS);
	
	private static Vector<String> sendedProcessorid = new Vector<String>() ;

	public static void sendFeedbackInfo(String processorid, String message,
			String component) throws CPSException {
		if(sendedProcessorid.contains(processorid)){
			return;
		}
		sendedProcessorid.add(processorid);
		ObjectNode node = JSONUtility.newObjectNode();
		node.put(MessageFields.PROCESSORID, processorid);
		node.put(MessageFields.FEEDBACK_INFO, message);
		node.put("Component", component);
		producer.send(MessageType.FEEDBACK_INFO.name(), node.toString()
				.getBytes());
	}

}
